package ds_industry_2025.ds.ds_06.T4


import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

import java.util.Properties

/*
   1、据Hudi的dwd_ds_hudi库中相关表或MySQL数据库shtd_store中订单相关表（order_detail、order_info、sku_info），对用户购买
   过的商品进行去重，将其转换为以下格式：第一列为用户id mapping，第二列为用户购买过的商品id mapping，按照user_id与sku_id进行
   升序排序，输出前5行，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下；
字段	类型	中文含义	备注
user_id	int	用户id的mapping对应键
sku_id	int	商品id的mapping对应键
提示：
Mapping操作：例如用户id：1、4、7、8、9，则做完mapping操作转为字典类型，键0对应用户id 1，键1对应用户id 4，以此类推
结果格式如下：
-------user_id_mapping与sku_id_mapping数据前5条如下：-------
0:0
0:89
1:1
1:2
1:3
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("特征工程01")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    //  导入隐式转换
    import spark.implicits._

    //  todo 读取需要用到的表的数据
    val sku_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "sku_info", conn)

    val order_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "order_info", conn)

    val detail_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "order_detail", conn)

    val user_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "user_info", conn)


    //  todo 剔除订单信息表(order_info)和订单详细表(order_detail)中的user_id和sku_id
    //   不存在现有维度表(sku_info和user_info)中的数据
    val user_id = user_info.select("id").withColumnRenamed("id","id_1").distinct()
    val sku_id = sku_info.select("id").withColumnRenamed("id","id_1").distinct()
    val order= order_info.join(user_id, user_id("id_1") === order_info("user_id"))
    val detail = detail_info.join(sku_id, sku_id("id_1") === detail_info("sku_id")).drop("id")


    //  todo 根据user_id和sku_id升序排列的方式使用排序函数转化为题目要求的格式
    //  todo rank:113  dense_rank:112   row_number:123
    //  下面之所以需要减1是因为要符合题目，开头为0，因为排序函数开头都是1
    order.join(detail,detail("order_id") === order("id"))
      .select("user_id","sku_id")
      .distinct()
      .withColumn(
        "user_id",
        dense_rank() over(Window.orderBy("user_id"))
      )
      .withColumn("user_id",col("user_id")-1)
      .withColumn(
        "sku_id",
        dense_rank() over(Window.orderBy("sku_id"))
      )
      .withColumn("sku_id",col("sku_id")-1)
      .orderBy("user_id")
      .limit(5)
      .foreach{
        r: Row => {
          println(r.get(0)+":"+r.get(1))
        }
      }


    spark.close()
  }

}
